/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Properties;

import jvs.peer.proxy.SocketRedirectThread;
import jvs.peer.util.Log;

import org.apache.commons.pool.impl.GenericKeyedObjectPool;


/**
 * @author qiangli
 * 
 */
public class SocketServer extends DataPacketService {

	private static final String TAG = "SocketServer";

	private static final int BUF_SIZE = 8192;
	
	private GenericKeyedObjectPool pool = new GenericKeyedObjectPool(
			new SocketPoolableObjectFactory(), 1);

	public class MyProxySocketThread extends SocketRedirectThread {

		private Socket out;

		public MyProxySocketThread(ServerSocket socket, Socket out) {
			super(socket);
			this.out = out;
		}

		@Override
		protected void tunnel(Socket in, Socket out) {
			try {
				InputStream is = in.getInputStream();

				receive(is);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		@Override
		protected void handle(Socket in) throws Exception {
			tunnel(in, out);
		}

	}

	private String host;
	private int port;

	private SocketRedirectThread listener;

	public SocketServer() {
		this.host = getLocalAddress();
	}

	private String getLocalAddress() {
		return "localhost";
	}

	@Override
	public Message receive() throws IOException {
		Message msg = super.receive();
		
		if (Log.DEBUG) {
			Log.d(TAG, this + " revd<<< " + msg);
		}
		
		return msg;
	}

	@Override
	public void init(Properties props) throws Exception {
		this.port = Integer.parseInt(props.getProperty("port"));
	}

	@Override
	public void send(Message msg) throws Exception {
		// fill in from
		msg.getHeader().setFrom(host + "/" + port);

		final String key =  msg.getHeader().getPeer();
		Socket out = (Socket) pool.borrowObject(key);
		
		try {
			OutputStream os = out.getOutputStream();

			send(msg, os);

			os.flush();
			
			if (Log.DEBUG) {
				Log.d(TAG, this + " sent>>> " + msg);
			}
		} catch (Exception e) {
			pool.invalidateObject(key, out);
	
			throw e;
		} finally {
			if (out != null) {
				pool.returnObject(key, out);
			}
		}
	}

	@Override
	public void startup() throws Exception {
		listener = new MyProxySocketThread(new ServerSocket(port), null);
		listener.start();
	}

	@Override
	public void shutdown() throws Exception {
		listener.interrupt();
		listener = null;
	}

	public int getBufferSize() {
		return BUF_SIZE;
	}
}
